package com.yanggu.flink.datastream_api.wordcount;


import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Objects;

public class StreamWordCount {

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.set(RestOptions.PORT, 8081);
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        //设置运行模式为流模式
        environment.setRuntimeMode(RuntimeExecutionMode.STREAMING);

        String path = Objects.requireNonNull(StreamWordCount.class.getResource("/hello.txt")).getPath();
        FileSource<String> fileSource = FileSource
                .forRecordStreamFormat(new TextLineInputFormat(), new Path(path))
                .build();

        environment
                .fromSource(fileSource, WatermarkStrategy.noWatermarks(), "file-monitor")
                .flatMap((value, out) -> {
                    for (String s : value.split(" ")) {
                        out.collect(s);
                    }
                }, Types.STRING)
                .map(value -> Tuple2.of(value, 1)).returns(Types.TUPLE(Types.STRING, Types.INT))
                .keyBy(tuple -> tuple.f0)
                .sum(1)
                .print("data print>>>");

        environment.execute("StreamWordCount");
    }

}
